package io.plus.interactive.collections;

import drds.common.$;
import io.plus.interactive.AsynchronousSocketChannelContext;
import io.plus.interactive.Configuration;
import io.plus.utils.lock.LockUtils;
import io.plus.utils.lock.ReadWriteLockMap;
import io.plus.utils.lock.ReadWriteLockSet;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock;

/**
 * 一对多  (token <--> AsynchronousSocketChannelContext)<br>
 */
@Slf4j
public class Tokens {

    private final Object lock = new Object();
    /**
     * key: token
     * value: ReadWriteLockSet<AsynchronousSocketChannelContext>
     */
    private ReadWriteLockMap<String, ReadWriteLockSet<AsynchronousSocketChannelContext>> tokenToChannelContextSetMap = new ReadWriteLockMap<>(new HashMap<String, ReadWriteLockSet<AsynchronousSocketChannelContext>>());

    /**
     * 绑定token.
     */
    public void bind(String token, AsynchronousSocketChannelContext asynchronousSocketChannelContext) {
        if (asynchronousSocketChannelContext.clientConfiguration.isShortConnection) {
            return;
        }

        if ($.isNullOrEmpty(token)) {
            return;
        }

        try {
            ReadWriteLockSet<AsynchronousSocketChannelContext> channelContextSet = tokenToChannelContextSetMap.get(token);
            if (channelContextSet == null) {
                LockUtils.runWriteOrWaitRead("_tokens_bind_" + token, lock, () -> {
                    if (tokenToChannelContextSetMap.get(token) == null) {
                        tokenToChannelContextSetMap.put(token, new ReadWriteLockSet<>(new HashSet<>()));
                    }
                });
                channelContextSet = tokenToChannelContextSetMap.get(token);
            }
            channelContextSet.add(asynchronousSocketChannelContext);

            asynchronousSocketChannelContext.setToken(token);
        } catch (Throwable e) {
            log.error("", e);
        } finally {
        }

    }

    public ReadWriteLockSet<AsynchronousSocketChannelContext> find(Configuration configuration, String token) {
        if (configuration.isShortConnection) {
            return null;
        }

        if ($.isNullOrEmpty(token)) {
            return null;
        }

        Lock lock = tokenToChannelContextSetMap.getReadLock();
        lock.lock();
        try {
            return tokenToChannelContextSetMap.getObject().get(token);
        } catch (Throwable e) {
            throw e;
        } finally {
            lock.unlock();
        }
    }

    public ReadWriteLockMap<String, ReadWriteLockSet<AsynchronousSocketChannelContext>> getTokenToChannelContextSetMap() {
        return tokenToChannelContextSetMap;
    }

    /**
     * 解除channelContext绑定的token
     */
    public void unbind(AsynchronousSocketChannelContext asynchronousSocketChannelContext) {
        if (asynchronousSocketChannelContext.clientConfiguration.isShortConnection) {
            return;
        }
        try {
            String token = asynchronousSocketChannelContext.getToken();
            if ($.isNullOrEmpty(token)) {
                log.debug("{}, {}, 并没有绑定Token", asynchronousSocketChannelContext.clientConfiguration.getName(), asynchronousSocketChannelContext.toString());
                return;
            }

            try {
                ReadWriteLockSet<AsynchronousSocketChannelContext> channelContextSet = tokenToChannelContextSetMap.get(token);
                if (channelContextSet == null) {
                    log.warn("{}, {}, token:{}, 没有找到对应的SetWithLock", asynchronousSocketChannelContext.clientConfiguration.getName(), asynchronousSocketChannelContext.toString(), token);
                    return;
                }
                asynchronousSocketChannelContext.setToken(null);
                channelContextSet.remove(asynchronousSocketChannelContext);

                if (channelContextSet.size() == 0) {
                    tokenToChannelContextSetMap.remove(token);
                }
            } catch (Throwable e) {
                throw e;
            }
        } catch (Throwable e) {
            log.error(e.toString(), e);
        }
    }


    public void unbind(Configuration configuration, String token) {
        if (configuration.isShortConnection) {
            return;
        }
        if ($.isNullOrEmpty(token)) {
            return;
        }

        try {
            ReadWriteLockSet<AsynchronousSocketChannelContext> channelContextSet = tokenToChannelContextSetMap.get(token);
            if (channelContextSet == null) {
                return;
            }

            WriteLock writeLock = channelContextSet.getWriteLock();
            writeLock.lock();
            try {
                Set<AsynchronousSocketChannelContext> $AsynchronousSocketChannelContextSet = channelContextSet.getObject();
                if ($AsynchronousSocketChannelContextSet.size() > 0) {
                    for (AsynchronousSocketChannelContext asynchronousSocketChannelContext : $AsynchronousSocketChannelContextSet) {
                        asynchronousSocketChannelContext.setToken(null);
                    }
                    $AsynchronousSocketChannelContextSet.clear();
                }

                tokenToChannelContextSetMap.remove(token);
            } catch (Throwable e) {
                log.error(e.getMessage(), e);
            } finally {
                writeLock.unlock();
            }
        } catch (Throwable e) {
            throw e;
        }
    }
}
